package com.atguigu.sparkcore.day02.kv

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Author atguigu
 * Date 2020/10/28 15:46
 */
object JoinDemo {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setAppName("JoinDemo").setMaster("local[2]")
        val sc: SparkContext = new SparkContext(conf)
        val list1 = List(30, 50, 70, 6, 10, 20, 30).map((_, "a"))
        val list2 = List(30, 70, 60, 1, 2, 30).map((_, "b"))
        val rdd1 = sc.parallelize(list1, 2)
        val rdd2 = sc.parallelize(list2, 2)
        
        //val rdd3: RDD[(Int, (String, String))] = rdd1.join(rdd2)
        //  val rdd3: RDD[(Int, (String, Option[String]))] = rdd1.leftOuterJoin(rdd2)
        //      val rdd3: RDD[(Int, (Option[String], String))] = rdd1.rightOuterJoin(rdd2)
        val rdd3: RDD[(Int, (Option[String], Option[String]))] = rdd1.fullOuterJoin(rdd2)
        
        // Option:  Some None
        rdd3.collect.foreach(println)
        sc.stop()
    }
}

/*
两个RDD进行连接

sql join
    1. join
           两边都有
    2. left join
            左边都有  右边用null补齐
    3. right join
    
    4. full join
            对方用null
            
     on x.id=y.id
     
val rdd4 =  rdd3.map{
            case (k, (v1, Some(v2))) =>
                (k, v1, v2)
            case (k, (v1, None)) =>
                (k, v1, "1000")
        }

 */